/**
 * 
 */
package com.dangdang.kafaka;

import java.util.HashMap;
import java.util.Map;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.MessageListener;

/**
 *
 * @author ligeng 2019年6月27日
 */
public class DefaultKafkaListenerContainer implements InitializingBean ,DisposableBean{

	private Map<String, Object> propsMap = new HashMap<>();
	private KafkaProperties kafkaProperties;

	private MessageListener<?, ?> messageListener;

	private String[] topics;

	private Integer  concurrency =  null ;
	
	private ConcurrentMessageListenerContainer<?, ?> factory;
	
	
	private Boolean start = false;

	@Override
	public void destroy() throws Exception {

		if(start && factory != null) {
			factory.stop();
		}
	}

	@Override
	public void afterPropertiesSet() throws Exception {
		propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getBootstrapServers());
		propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,kafkaProperties.getAutoCommit() );
		propsMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,kafkaProperties.getInterval()== null ? "100":kafkaProperties.getInterval());
		propsMap.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, kafkaProperties.getSessionTimeout()== null ?"15000" :kafkaProperties.getSessionTimeout());
		propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
		propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
		propsMap.put(ConsumerConfig.GROUP_ID_CONFIG,kafkaProperties.getGroupId() == null ?  "group1" :kafkaProperties.getGroupId() );
		propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, kafkaProperties.getReset() == null ?  "earliest" :kafkaProperties.getReset()  );
		DefaultKafkaConsumerFactory<String, String> defaultKafkaConsumerFactory = new DefaultKafkaConsumerFactory<>(
				propsMap);
		ContainerProperties containerProperties = new ContainerProperties(topics);
		containerProperties.setMessageListener(messageListener);
		factory = new ConcurrentMessageListenerContainer<>(
				defaultKafkaConsumerFactory, containerProperties);
		factory.setConcurrency(concurrency == null ?  kafkaProperties.getConsumerThreads() : concurrency);
		factory.start();
		start = true;
	}

	public void setupMessageListener(MessageListener<?, ?> messageListener) {
		this.messageListener = messageListener;
	}

	public void start() {
		if(!start && factory != null) {
			factory.start();
		}
	}

	/**
	 * @return the kafkaProperties
	 */
	public KafkaProperties getKafkaProperties() {
		return kafkaProperties;
	}

	/**
	 * @param kafkaProperties the kafkaProperties to set
	 */
	public void setKafkaProperties(KafkaProperties kafkaProperties) {
		this.kafkaProperties = kafkaProperties;
	}

	/**
	 * @return the messageListener
	 */
	public MessageListener<?, ?> getMessageListener() {
		return messageListener;
	}

	/**
	 * @param messageListener the messageListener to set
	 */
	public void setMessageListener(MessageListener<?, ?> messageListener) {
		this.messageListener = messageListener;
	}

	/**
	 * @return the topics
	 */
	public String[] getTopics() {
		return topics;
	}

	/**
	 * @param topics the topics to set
	 */
	public void setTopics(String[] topics) {
		this.topics = topics;
	}

	/**
	 * @return the concurrency
	 */
	public Integer getConcurrency() {
		return concurrency;
	}

	/**
	 * @param concurrency the concurrency to set
	 */
	public void setConcurrency(Integer concurrency) {
		this.concurrency = concurrency;
	}

	/**
	 * @return the start
	 */
	public Boolean isStart() {
		return start;
	}

	
}
